package com.xiguanlezz.cn.client.naming.remote.grpc.redo;

import com.xiguanlezz.cn.api.PropertyKeyConst;
import com.xiguanlezz.cn.api.common.Constants;
import com.xiguanlezz.cn.api.naming.pojo.Instance;
import com.xiguanlezz.cn.api.naming.pojo.ServiceInfo;
import com.xiguanlezz.cn.api.naming.utils.NamingUtils;
import com.xiguanlezz.cn.client.env.NacosClientProperties;
import com.xiguanlezz.cn.client.naming.remote.grpc.NamingGrpcClientProxy;
import com.xiguanlezz.cn.client.naming.remote.grpc.redo.data.InstanceRedoData;
import com.xiguanlezz.cn.client.naming.remote.grpc.redo.data.SubscriberRedoData;
import com.xiguanlezz.cn.common.executor.NameThreadFactory;
import com.xiguanlezz.cn.common.remote.client.Connection;
import com.xiguanlezz.cn.common.remote.client.ConnectionEventListener;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.*;

/**
 * @Author：习惯了种种
 * @CreateDate：2024/11/25 15:14
 * @Description：失败重试对象
 */
public class NamingGrpcRedoService implements ConnectionEventListener {
    // 执行重试操作的定时任务执行器
    private final ScheduledExecutorService redoExecutor;
    // 重试操作的延迟时间，就是提交定时任务时设定的延迟时间
    private long redoDelayTime;
    // 执行重试操作的线程名称
    private static final String REDO_THREAD_NAME = "com.alibaba.nacos.client.naming.grpc.redo";
    // 执行重试操作的线程数量
    private int redoThreadCount;

    // 存放注册操作的重试对象的map
    private final ConcurrentMap<String, InstanceRedoData> registeredInstances = new ConcurrentHashMap<>();
    // 存放订阅实例的重试对象的map
    private final ConcurrentMap<String, SubscriberRedoData> subscribes = new ConcurrentHashMap<>();


    //表示当前客户端和服务端是否成功建立了连接
    private volatile boolean connected = false;

    public NamingGrpcRedoService(NamingGrpcClientProxy clientProxy, NacosClientProperties properties) {
        setProperties(properties);
        redoExecutor = new ScheduledThreadPoolExecutor(redoThreadCount, new NameThreadFactory(REDO_THREAD_NAME));
        // 定时任务默认执行的间隔是3秒，请求发送的超时时间也是3秒
        redoExecutor.scheduleWithFixedDelay(
                new RedoScheduledTask(clientProxy, this),
                redoDelayTime,
                redoDelayTime,
                TimeUnit.MILLISECONDS);
    }

    // 为成员变量赋值的方法
    private void setProperties(NacosClientProperties properties) {
        redoDelayTime = properties.getLong(PropertyKeyConst.REDO_DELAY_TIME, Constants.DEFAULT_REDO_DELAY_TIME);
        redoThreadCount = properties.getInteger(PropertyKeyConst.REDO_DELAY_THREAD_COUNT,
                Constants.DEFAULT_REDO_THREAD_COUNT);
    }

    /**
     * 缓存服务注册的重试对象
     */
    public void cacheInstanceForRedo(String serviceName, String groupName, Instance instance) {
        String key = NamingUtils.getGroupedName(serviceName, groupName);
        InstanceRedoData redoData = InstanceRedoData.build(serviceName, groupName, instance);
        synchronized (registeredInstances) {
            registeredInstances.put(key, redoData);
        }
    }

    /**
     * 移除服务注册的重试对象
     */
    public void removeInstanceForRedo(String serviceName, String groupName) {
        String key = NamingUtils.getGroupedName(serviceName, groupName);
        synchronized (registeredInstances) {
            InstanceRedoData redoData = registeredInstances.get(key);
            if (null != redoData && !redoData.isExpectedRegistered()) {
                registeredInstances.remove(key);
            }
        }
    }

    /**
     * 当服务注册成功后，调用该方法将重试对象设置为已注册状态
     */
    public void instanceRegistered(String serviceName, String groupName) {
        String key = NamingUtils.getGroupedName(serviceName, groupName);
        synchronized (registeredInstances) {
            InstanceRedoData redoData = registeredInstances.get(key);
            if (null != redoData) {
                redoData.registered();
            }
        }
    }

    /**
     * 刚开始注销服务时，调用该方法将重试对象期望状态改为注销，同时设置unregistering为true
     */
    public void instanceDeregister(String serviceName, String groupName) {
        String key = NamingUtils.getGroupedName(serviceName, groupName);
        synchronized (registeredInstances) {
            InstanceRedoData redoData = registeredInstances.get(key);
            if (null != redoData) {
                redoData.setUnregistering(true);
                redoData.setExpectedRegistered(false);
            }
        }
    }

    /**
     * 当服务注销成功后，调用该方法将重试对象设置为已注销状态
     */
    public void instanceDeregistered(String serviceName, String groupName) {
        String key = NamingUtils.getGroupedName(serviceName, groupName);
        synchronized (registeredInstances) {
            InstanceRedoData redoData = registeredInstances.get(key);
            if (null != redoData) {
                redoData.unregistered();
            }
        }
    }

    /**
     * 得到所有在服务注册过程中需要重试的重试对象列表
     */
    public Set<InstanceRedoData> findInstanceRedoData() {
        Set<InstanceRedoData> result = new HashSet<>();
        synchronized (registeredInstances) {
            for (InstanceRedoData each : registeredInstances.values()) {
                if (each.isNeedRedo()) {
                    result.add(each);
                }
            }
        }
        return result;
    }

    /**
     * 缓存订阅实例的重试对象
     */
    public void cacheSubscriberForRedo(String serviceName, String groupName, String cluster) {
        String serviceKey = ServiceInfo.getKey(NamingUtils.getGroupedName(serviceName, groupName), cluster);
        SubscriberRedoData redoData = SubscriberRedoData.build(serviceName, groupName, cluster);
        synchronized (subscribes) {
            subscribes.put(serviceKey, redoData);
        }
    }

    /**
     * 移除订阅实例的重试对象
     */
    public void removeSubscriberForRedo(String serviceName, String groupName, String cluster) {
        String serviceKey = ServiceInfo.getKey(NamingUtils.getGroupedName(serviceName, groupName), cluster);
        synchronized (subscribes) {
            SubscriberRedoData redoData = subscribes.get(serviceKey);
            if (null != redoData && !redoData.isExpectedRegistered()) {
                subscribes.remove(serviceKey);
            }
        }
    }

    /**
     * 当服务端完成订阅服务实例后，调用该方法将重试对象设置为已订阅状态
     */
    public void subscriberRegistered(String serviceName, String groupName, String cluster) {
        String serviceKey = ServiceInfo.getKey(NamingUtils.getGroupedName(serviceName, groupName), cluster);
        synchronized (subscribes) {
            SubscriberRedoData redoData = subscribes.get(serviceKey);
            if (null != redoData) {
                redoData.setRegistered(true);
            }
        }
    }

    /**
     * 刚开始取消订阅服务实例时，调用该方法将重试对象期望状态改为取消订阅，同时设置unregistering为true
     */
    public void subscriberDeregister(String serviceName, String groupName, String cluster) {
        String serviceKey = ServiceInfo.getKey(NamingUtils.getGroupedName(serviceName, groupName), cluster);
        synchronized (subscribes) {
            SubscriberRedoData redoData = subscribes.get(serviceKey);
            if (null != redoData) {
                redoData.setUnregistering(true);
                redoData.setExpectedRegistered(false);
            }
        }
    }

    /**
     * 得到所有在订阅实例过程中需要重试的重试对象列表
     */
    public Set<SubscriberRedoData> findSubscriberRedoData() {
        Set<SubscriberRedoData> result = new HashSet<>();
        synchronized (subscribes) {
            for (SubscriberRedoData each : subscribes.values()) {
                if (each.isNeedRedo()) {
                    result.add(each);
                }
            }
        }
        return result;
    }

    /**
     * 查看实例是否已经被订阅
     */
    public boolean isSubscriberRegistered(String serviceName, String groupName, String cluster) {
        String serviceKey = ServiceInfo.getKey(NamingUtils.getGroupedName(serviceName, groupName), cluster);
        synchronized (subscribes) {
            SubscriberRedoData redoData = subscribes.get(serviceKey);
            return null != redoData && redoData.isRegistered();
        }
    }

    @Override
    public void onConnected(Connection connection) {
        connected = true;
    }

    @Override
    public void onDisConnect(Connection connection) {
        connected = false;
        // 因为连接断开了，后面会进行重连，
        // Nacos服务端在客户端挂掉时，只会将Client对象注销，如果连接重连成功不重新注册服务实例，则客户端持有老的ClientId无法订阅到别的服务信息
        // 这里手动将registered设置为false，等客户端重新连接了服务端之后，会将服务实例重新注册到服务端
        registeredInstances.values().forEach(instanceRedoData -> instanceRedoData.setRegistered(false));
    }
}
